package org.example.zookeeper.zk;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ZookeeperRegistryCenter {

    private CuratorFramework client;

    private final Map<String, NodeCache> nodeCacheMap = new ConcurrentHashMap<>();

    public void init() {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);

        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("datasource")  // 独立的命名空间 /base
                .build();

        client.start();
    }

    public void close() throws Exception{
        for (Map.Entry<String, NodeCache> nodeCache : nodeCacheMap.entrySet()) {
            nodeCache.getValue().close();
        }
        waitForCacheClose();
        CloseableUtils.closeQuietly(client);
    }

    /*
     *  // TODO
     * sleep 500ms, let cache client close first and then client, otherwise will throw exception
     * reference：https://issues.apache.org/jira/browse/CURATOR-157
     */
    private void waitForCacheClose() {
        try {
            Thread.sleep(500L);
        } catch (final InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

    public String getDirectly(final String key) throws Exception {
        byte[] bytes = client.getData().forPath(key);
        return new String(bytes, "UTF-8");
    }

    public boolean isExisted(final String key) throws Exception{
        return null != client.checkExists().forPath(key);
    }

    public void persist(final String key) throws Exception{
        if (!isExisted(key)) {
            client.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(key);
        }
    }

    public void persistEphemeral(final String key) throws Exception{
        client.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key);
    }

    public List<String> getChildren(final String key) throws Exception{
        List<String> children = client.getChildren().forPath(key);
        return children;
    }

    public void addCacheData(final String cachePath) throws Exception{
        NodeCache nodeCache = new NodeCache(client, cachePath, false);
        nodeCache.start();
        nodeCacheMap.put(cachePath, nodeCache);
    }

    public NodeCache getNodeCache(final String cachePath) {
        return nodeCacheMap.get(cachePath);
    }
}

